/*
 *  Copyright 2015 PayPal
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.squbs.bottlesvc

import akka.actor.Props
import akka.event.slf4j.SLF4JLogging
import akka.http.scaladsl.marshalling.ToResponseMarshallable
import akka.http.scaladsl.model.{HttpEntity, _}
import akka.http.scaladsl.server.Route
import akka.stream.actor.ActorPublisher
import akka.stream.scaladsl.Source
import akka.util.ByteString
import org.squbs.bottlemsgs._
import org.squbs.unicomplex.{RouteDefinition, WebContext}

// this class defines our service behavior independently from the service actor
class BottleSvc extends RouteDefinition with WebContext {
  
  def route: Route =
    path("hello") {
      get {
        val rs = HttpEntity(ContentTypes.`text/html(UTF-8)`, s"""
            <html>
              <body>
                <h1>Say hello to <i>spray-routing</i> on <i>spray-can</i>!</h1>
              </body>
            </html>""")
        complete(ToResponseMarshallable(rs))
      }
    } ~
    path(""".*\.html""".r) { name => getFromResource("html/" + name)
    } ~  
    path("events") {
      get {
        val data = Source.actorPublisher[ByteString](Props(classOf[OnePublisher]))
        val c = ContentType(MediaType.text("event-stream"), HttpCharsets.`UTF-8`)
        println(s"c == $c")
        val rs = HttpEntity(c, data)
        complete(ToResponseMarshallable(rs))
      }
    }
  }

class OnePublisher extends ActorPublisher[ByteString] with SLF4JLogging {
  import akka.stream.actor.ActorPublisherMessage._
  context.actorSelection("/user/bottlecube/lyrics") ! StartEvents
  def toSSE(msg: String): String = "event: lyric\ndata: " + msg.replace("\n", "\ndata: ") + "\n\n"
  val streamEnd = "event: streamEnd\ndata: End of stream\n\n"
  var buf = Vector.empty[String]
  def processEvent(eventMessage: String): Unit = {
    log.info('\n' + eventMessage)
    if (buf.isEmpty && totalDemand > 0)
      onNext(ByteString(eventMessage))
    else {
      buf :+= eventMessage
      deliverBuf()
    }
  }
  override def receive: Receive = {
        case Event(msg) =>
          processEvent(toSSE(msg))
        case EndEvents  =>
          processEvent(streamEnd)
          context.stop(self)
    case Request(r) =>
      log.info(s"Request: $r, $totalDemand")
      deliverBuf()
    case Cancel =>
      context stop self
  }
  final def deliverBuf(): Unit = {
    if (totalDemand > 0) {
      if (totalDemand <= Int.MaxValue) {
        val (use, keep) = buf.splitAt(totalDemand.toInt)
        buf = keep
        use foreach { str =>
          onNext(ByteString(str))
        }
      } else {
        val (use, keep) = buf.splitAt(totalDemand.toInt)
        buf = keep
        use foreach  { str =>
          onNext(ByteString(str))
        }
        deliverBuf()
      }
    }
  }
}
